package com.xuecheng.media.service.jobhandler;

import com.xuecheng.base.utils.Mp4VideoUtil;
import com.xuecheng.media.mapper.MediaFilesMapper;
import com.xuecheng.media.model.po.MediaFiles;
import com.xuecheng.media.model.po.MediaProcess;
import com.xuecheng.media.service.MediaFileProcessService;
import com.xuecheng.media.service.MediaFileService;
import com.xuecheng.media.service.impl.MediaFileServiceImpl;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author: Hes
 * @description 视频转码任务
 * @date: 2023/6/22 21:54
 */
@Slf4j
@Component
public class VideoTask {
    @Value("${videoprocess.ffmpegpath}")
    private String ffmpegpath;

    @Autowired
    MediaFileProcessService mediaFileProcessService;

    @Autowired
    MediaFileService mediaFileService;


    /**
     * 分片广播处理视频转码
     * 1、任务调度中心广播作业分片。
     * 2、执行器收到广播作业分片，从数据库读取待处理任务，读取未处理及处理失败的任务。
     * 3、执行器更新任务为处理中，根据任务内容从MinIO下载要处理的文件。
     * 4、执行器启动多线程去处理任务。
     * 5、任务处理完成，上传处理后的视频到MinIO。
     * 6、将更新任务处理结果，如果视频处理完成除了更新任务处理结果以外还要将文件的访问地址更新至任务处理表及文件表中，最后将任务完成记录写入历史表。
     *
     * @throws Exception
     */
    @XxlJob("videoJobHandler")
    public void videoJobHandler() {

        // 分片参数
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();
        //cpu核数
        int processors = Runtime.getRuntime().availableProcessors();
        //查询待处理任务
        List<MediaProcess> mediaProcessList = mediaFileProcessService.getMediaProcessList(shardIndex, shardTotal, processors);
        int size = mediaProcessList.size();
        if (size < 0) {
            return;
        }
        log.debug("取出待处理视频任务{}条", size);
        // 启动和任务个数相同的线程的线程池
//        ExecutorService executorServicePool = Executors.newFixedThreadPool(size);
//        ThreadPoolExecutor executorServicePool = new ThreadPoolExecutor(核心线程数,
//        最大线程数，
//        临时线程存活时间，
//        存活时间单位，
//        线程阻塞队列，
//        线程工厂按需创建线程，
//        当没有更多线程或队列插槽可用时的处理程序)
//        ThreadPoolExecutor executorServicePool = new ThreadPoolExecutor(
//                7,
//                9,
//                0L,
//                TimeUnit.SECONDS,
//                new LinkedBlockingDeque<>(5),
//                new ThreadFactory() {
//            @Override
//            public Thread newThread(@NotNull Runnable r) {
//                Thread thread = new Thread("videoTask:");
//                return thread;
//            }
//        },new ThreadPoolExecutor.AbortPolicy());

        //可简写
        ThreadPoolExecutor executorServicePool = new ThreadPoolExecutor(
                7,
                9,
                0L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(5),
                r -> new Thread("videoTask:"),
                new ThreadPoolExecutor.AbortPolicy());
        //计数器
        CountDownLatch countDownLatch = new CountDownLatch(size);
        //将处理任务加入线程池
        for (MediaProcess mediaProcess : mediaProcessList) {
            executorServicePool.execute(() -> {
                Long taskId = mediaProcess.getId();
                //通过乐观锁抢占任务,执行器更新任务为处理中
                boolean b = mediaFileProcessService.startTask(taskId);
                if (!b) {
                    return;
                }
                log.debug("开始执行任务:{}", mediaProcess);
                //根据任务内容从MinIO下载要处理的文件
                String filename = mediaProcess.getFilename();
                String extension = filename.substring(filename.lastIndexOf("."));
                String fileId = mediaProcess.getFileId();
                String bucket = mediaProcess.getBucket();
                File originalFile = mediaFileService.downloadFileFromMinIO(fileId, extension, bucket);
                if (originalFile == null) {
                    log.debug("下载待处理文件失败,originalFile:{}", mediaProcess.getBucket().concat(mediaProcess.getFilePath()));
                    mediaFileProcessService.saveProcessFinishStatus(taskId, "3", fileId, "", "下载待处理文件失败");
                    return;
                }
                File mp4File = null;
                try {
                    mp4File = File.createTempFile("mp4", ".mp4");
                } catch (IOException e) {
                    e.printStackTrace();
                    log.error("创建mp4临时文件失败");
                    mediaFileProcessService.saveProcessFinishStatus(taskId, "3", fileId, "", "创建mp4临时文件失败");
                    return;
                }
                //开始处理视频
                Mp4VideoUtil videoUtil = new Mp4VideoUtil(ffmpegpath, originalFile.getAbsolutePath(), mp4File.getName(), mp4File.getAbsolutePath());
                //开始视频转换，成功将返回success
                String result = videoUtil.generateMp4();
                if (StringUtils.isEmpty(result) || !result.equals("success")) {
                    log.error("视频转码失败");
                    mediaFileProcessService.saveProcessFinishStatus(taskId, "3", fileId, "", result);
                    return;
                }
                //上传minio
                String absolutePath = mp4File.getAbsolutePath();
                String filePathByMd5 = getFilePathByMd5(fileId, ".mp4");
                String objectName = "/" + bucket + "/" + filePathByMd5;
                try {
                    boolean res = mediaFileService.addMediaFilesToMinIO(bucket, "video/mp4", objectName, absolutePath);
                    if (!res) {
                        log.error("上传minio失败");
                        mediaFileProcessService.saveProcessFinishStatus(taskId, "3", fileId, "", "上传minio失败");
                        return;
                    }
                    //视频处理成功
                    String url = objectName;
                    mediaFileProcessService.saveProcessFinishStatus(taskId, "2", fileId, url, "");
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("上传视频失败或入库失败,视频地址:{},错误信息:{}", bucket + objectName, e.getMessage());
                    mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, "处理后视频上传或入库失败");
                } finally {
                    countDownLatch.countDown();
                }

            });

        }
        //等待,给一个充裕的超时时间,防止无限等待，到达超时时间还没有处理完成则结束任务
        try {
            countDownLatch.await(30, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

    }

    private String getFilePathByMd5(String fileMd5, String extension) {
        return fileMd5.substring(0, 1) + "/" + fileMd5.substring(1, 2) + "/" + fileMd5 + "/" + fileMd5 + extension;
    }
}
